package logsvc

import (
	"context"
	"github.com/olivere/elastic/v7"
	corev1 "gluenet/internal/agent/api/core/v1"
	"gluenet/internal/agent/depend"
	"gluenet/internal/agent/service/essvc"
	"gluenet/internal/glue-datasyncer/logs"
	"gluenet/pkg/ops"
	"gluenet/pkg/ops/impl/es"
	"gluenet/utils/rpc/v2/define"
	"reflect"
	"strconv"
	"strings"
	"time"
)

func (svc *Service) LogCounts(logCounts define.LogCounts) corev1.CommonResp {
	dbEs, err := es.NewEs(context.Background(), depend.Produce[*essvc.EsConfig]().Addr)
	if err != nil {
		svc.logger.Errorf(err.Error())
		return corev1.NewCommonFailureResp(err.Error())
	}
	client := dbEs.Client
	//1.Query from db
	//Set criteria for the query
	boolQuery := elastic.NewBoolQuery()
	termQueryGUID := elastic.NewQueryStringQuery("application_guid:" + logCounts.Guid)
	termQueryContainer := elastic.NewQueryStringQuery("container:" + logCounts.Container)
	rangeQueryTimeStamp := elastic.NewRangeQuery("timestamp").Gte(logCounts.Start).Lte(logCounts.End)
	boolQuery.Must(termQueryGUID, termQueryContainer, rangeQueryTimeStamp)
	//Set the aggregation with interval of timestamp and term of level
	//ParseDuration does not support the format like "1d"
	if strings.Contains(logCounts.Bucket, "d") {
		count, _ := strconv.Atoi(logCounts.Bucket[:strings.Index(logCounts.Bucket, "d")])
		hour := strconv.Itoa(count * 24)
		logCounts.Bucket = hour + "h"
	}
	interval, _ := time.ParseDuration(logCounts.Bucket)
	timeStamp := elastic.NewHistogramAggregation().Field("timestamp").Interval(interval.Seconds()).MinDocCount(0)
	histogram := elastic.NewTermsAggregation().Field("level.keyword").MinDocCount(0)
	timeStamp.SubAggregation("level", histogram)
	//Start query
	searchResult, err := client.Search().
		Index(ops.EsLog).
		Query(boolQuery).
		Aggregation("timestamp", timeStamp).
		Size(0).
		Do(context.Background())
	if err != nil {
		svc.logger.Errorf(err.Error())
		return corev1.NewCommonFailureResp(err.Error())
	}
	//2.Data organization
	aggResByTimeStamp, found := searchResult.Aggregations.Histogram("timestamp")
	if !found {
		svc.logger.Errorf(err.Error())
		return corev1.NewCommonFailureResp(err.Error())
	}
	var res []map[string]interface{}
	for _, bucket := range aggResByTimeStamp.Buckets {
		bucketValue := bucket.Key
		aggResByLevel, found := bucket.Aggregations.Terms("level")
		if !found {
			svc.logger.Errorf(err.Error())
			return corev1.NewCommonFailureResp(err.Error())
		}
		for _, log := range aggResByLevel.Buckets {
			res = append(res, map[string]interface{}{
				"counts": log.DocCount,
				"label":  "logs",
				"start":  int(bucketValue),
				"stop":   int(bucketValue + interval.Seconds()),
				"type":   log.Key,
			})
		}
	}
	//3.Response
	return corev1.NewCommonSuccessRespWithData(res)
}

func (svc *Service) LogInfo(logInfo define.LogInfo) corev1.CommonResp {
	dbEs, err := es.NewEs(context.Background(), depend.Produce[*essvc.EsConfig]().Addr)
	if err != nil {
		svc.logger.Errorf(err.Error())
		return corev1.NewCommonFailureResp(err.Error())
	}
	client := dbEs.Client
	//1.Query from db
	//Set criteria for the query
	boolQuery := elastic.NewBoolQuery()
	termQueryGUID := elastic.NewQueryStringQuery("application_guid:" + logInfo.Guid)
	termQueryContainer := elastic.NewQueryStringQuery("container:" + logInfo.Container)
	rangeQueryTimeStamp := elastic.NewRangeQuery("timestamp").Gte(logInfo.Start).Lte(logInfo.End)
	boolQuery.Must(termQueryGUID, termQueryContainer, rangeQueryTimeStamp)
	//Start query
	searchResult, err := client.Search().
		Index(ops.EsLog).
		Query(boolQuery).
		Sort("timestamp", true).
		Do(context.Background())
	if err != nil {
		svc.logger.Errorf(err.Error())
		return corev1.NewCommonFailureResp(err.Error())
	}
	//2.Data organization
	var res []map[string]interface{}
	if searchResult.TotalHits() > 0 {
		var esLog logs.EsLog
		for _, item := range searchResult.Each(reflect.TypeOf(esLog)) {
			if t, ok := item.(logs.EsLog); ok {
				res = append(res, map[string]interface{}{
					"label":   "logs",
					"message": t.Message,
					"time":    t.Timestamp,
					"type":    t.Level,
				})
			}
		}
	}
	//3.Response
	return corev1.NewCommonSuccessRespWithData(res)
}
